package com.example.mq.mqClient;

import com.example.mq.common.BasicArguments;
import com.example.mq.common.BasicReturns;
import com.example.mq.common.Request;
import com.example.mq.common.Response;
import com.example.mq.common.basic.SubScribeReturn;
import com.example.mq.mqServer.exception.MqException;
import com.example.mq.mqServer.util.SerializableUtils;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.couchbase.CouchbaseProperties;
import org.springframework.util.backoff.ExponentialBackOff;

import java.io.*;
import java.net.Socket;
import java.net.SocketException;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @BelongsProject: 0.mq
 * @BelongsPackage: com.example.mq.mqClient
 * @Author : 游在海里的鱼
 * @CreateTime : 2024-01-23  09:53
 * @Version : 1.0
 * @Description : TODO
 */
@Data
@Slf4j
public class Connection {
    private volatile Socket socket;
    //频道列表，管理一个TCP链接下的多个频道
    private ConcurrentHashMap<String,Channel> channelMap = new ConcurrentHashMap<>();

    //流
    private InputStream inputStream;
    private OutputStream outputStream;
    private DataInputStream dataInputStream;
    private DataOutputStream dataOutputStream;

    //扫描线程
    private Thread t;
    private volatile boolean isRunning = true;
    //线程池
    private ExecutorService callBackPool;

    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 构造方法
     * @Describe: TODO
     * @modifier: [public] [实例方法]
    **/
    public Connection(String host, int port){
        try {
            socket = new Socket(host,port);
            inputStream = socket.getInputStream();
            outputStream = socket.getOutputStream();
            dataInputStream = new DataInputStream(inputStream);
            dataOutputStream = new DataOutputStream(outputStream);

            callBackPool = Executors.newFixedThreadPool(4);

            //创建一个扫描线程,用于持续获取服务端响应
            t = new Thread(()->{
            while (isRunning){
                Response response = null;
                if(!socket.isClosed()){
                    response = readResponse();
                }
                if(response != null){
                    dispatchResponse(response);
                }
            }
            });
            t.start();
            log.info("[构造方法] [IP:"+socket.getInetAddress()+"的客户端]已经与服务端建立连接");
        }catch (IOException e) {
            log.error("[构造方法] [IP:"+host+"][端口:"+port+"]的socket创建失败");
            e.printStackTrace();
        }
    }

    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 释放资源
     * @Describe: TODD
     * @modifier: [public] [实例方法]
    **/
    public void close(){
        try {
            isRunning = false;
            socket.close();
            callBackPool.shutdown();
            channelMap = null;
            dataInputStream.close();
            dataOutputStream.close();
            inputStream.close();
            outputStream.close();
            log.info("[close] [IP:"+socket.getInetAddress().toString()+"的客户端]已经释放");
        } catch (IOException e) {
            log.error("[close] [IP:"+socket.getInetAddress().toString()+"的客户端]释放失败");
            e.printStackTrace();
        }
    }

    /* ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 用于区分响应,是服务器推送回的数据,还是对客户端请求作出的回应
     * @Describe: TODO
     * @modifier: [private] [实例方法]
    **/
    private void dispatchResponse(Response response) {
        try {
            //1.0xc代表服务端推送回的消息
            if(response.getType() == 0xc){
                //1.获取响应中的参数
                SubScribeReturn subScribeReturn = (SubScribeReturn) SerializableUtils
                        .fromBytes(response.getPayload());
                //2.判断channel是否存在，参数中指定的channel
                Channel channel = channelMap.get(subScribeReturn.getChannelId());
                if(channel == null){
                    new MqException("[Connection] [dispatchResponse] 连接中没有[响应"+subScribeReturn.getRid()
                            +"]的[channel"+subScribeReturn.getChannelId()+"]").printStackTrace();
                }
                //3.调用channel中的回调来处理消息，服务端推送回来的，使用线程池来处理
                callBackPool.submit(()->{
                    channel.getConsumer().handleDelivery(subScribeReturn.getConsumerTag()
                            , subScribeReturn.getBasicProperties(), subScribeReturn.getBody());
                });
            //2. 服务端对客户端请求作出的回应
            }else {
                //解析响应
                BasicReturns basicReturns = (BasicReturns) SerializableUtils
                        .fromBytes(response.getPayload());
                //获取channel，从响应提供的channel编号
                Channel channel = channelMap.get(basicReturns.getChannelId());
                if(channel == null){
                    new MqException("[Connection] [dispatchResponse] 连接中没有[响应"+basicReturns.getRid()
                            +"]的[channel"+basicReturns.getChannelId()+"]").printStackTrace();
                }
                //将响应交给对应的channel
                channel.putReturn(basicReturns);
            }
        } catch (IOException | ClassNotFoundException e) {
            e.printStackTrace();
        }
    }


    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 读取响应
     * @Describe: TODO
     * @modifier: [public] [实例方法]
    **/
    public Response readResponse(){
        try {
            Response response = new Response();
            response.setType(dataInputStream.readInt());
            response.setLength(dataInputStream.readInt());
            byte[] payload = new byte[response.getLength()];
            int count = dataInputStream.read(payload);
            if(count != response.getLength()){
                new MqException("[Connection] [readResponse] 流操作出现异常").printStackTrace();
            }
            response.setPayload(payload);
            return response;
        }catch (EOFException | SocketException e){

        }catch (IOException e) {
            log.error("[readResponse] 流操作出现异常");
            e.printStackTrace();
        }
        return null;
    }

    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 创建channel
     * @Describe: TODO
     * @modifier: [public] [实例方法]
    **/
    public Channel createChannel(){
        //1.在客户端创建channel
        String channelId = "C-"+ UUID.randomUUID();
        Channel channel = new Channel(channelId,this);
        channelMap.put(channelId,channel);
        //2.在服务端创建channel
        if(!channel.createChannel()){
            channelMap.remove(channelId);
            log.error("[createChannel] 创建chanel失败");
            return null;
        }
        return channel;
    }

    /** ---------------------------------------------------------------------------------------------------------------------
     * @Method  : 发送请求
     * @Describe: TODO
     * @modifier: [public] [实例方法]
    **/
    public void writeRequest(Request request){
        try {
            dataOutputStream.writeInt(request.getType());
            dataOutputStream.writeInt(request.getLength());
            dataOutputStream.write(request.getPayload());
            dataOutputStream.flush();
        } catch (IOException e) {
            log.error("[writeRequest] 流操作出现异常");
            e.printStackTrace();
        }
    }
    
}
